package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * Created by Smexy on 2023/4/8
 */
public class Demo7_WindowJoin
{
    public static void main(String[] args) throws Exception {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs())
             ;

        //两个流
        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);


        SingleOutputStreamOperator<WaterSensor> ds2 = env
            .socketTextStream("hadoop102", 8889)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);

        //windowJoin，先window，再join。把进入同一个窗口的两个流中的数据，按照key进行关联。两个流进行Join(内连接，交集)
        ds1.join(ds2)
           //指定关联条件  t1 join t2 on t1.key = t2.key
           .where(WaterSensor::getId)
           .equalTo(WaterSensor::getId)
           .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            //符合条件的数据会自动Join，只需要编写函数处理Join后的数据
           // apply: 非滚动聚合。窗口触发运算时只调用一次
           .apply(new JoinFunction<WaterSensor, WaterSensor, String>()
           {
               @Override
               public String join(WaterSensor first, WaterSensor second) throws Exception {
                   return first + "===" +second;
               }
           })
           .print();

        env.execute();

    }
}
